package net.oschina.simpleplus;

import java.util.concurrent.*;

/**
 * @author ; simpleplus(simpleplus@sina.cn)
 */
public class ActionExecutor {

    private final ThreadPoolExecutor executor;

    private final String name;

    private final ActionQueue defaultQueue;

    private DelayedThread delayedThread;

    private boolean runing = true;


    public ActionExecutor(final int corePoolSize, final int maxPoolSize, final String name){
        this.name = name == null ? "customer" : name;
        //超出corePoolSize数量之后的线程 超过5分钟空闲将被回收
        int keepAliveTime = 5;
        TimeUnit unit = TimeUnit.MINUTES;
        final LinkedTransferQueue<Runnable> workQueue = new LinkedTransferQueue<Runnable>();
        final RejectedExecutionHandler handler = new RejectedExecutionHandler() {
            @Override
            public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
                workQueue.offer(r);
            }
        };

        executor = new ThreadPoolExecutor(corePoolSize, maxPoolSize, keepAliveTime, unit, workQueue, handler);

        defaultQueue = new ActionQueue(this);

    }

    public void execute(Action next) {
        executor.execute(next);
    }

    public void checkinDelayedAction(DelayedAction delayedAction) {
        if(delayedThread == null){
            configAndCreateDelayedThread();
        }
        delayedThread.checkinDelayAction(delayedAction);
    }

    private synchronized void configAndCreateDelayedThread() {
        if (delayedThread == null){
            delayedThread = new DelayedThread();
            delayedThread.start();
        }
    }


    public void stop(){
        if(runing){
            runing = false;
            if(executor != null){
                executor.shutdown();
            }
            if(delayedThread != null){
                delayedThread.forceStop();
                delayedThread = null;
            }
        }

    }


    private static class DelayedThread extends Thread{
        private boolean stop;

        private ConcurrentLinkedDeque<DelayedAction> queue;

        private static final long ZZ_TIME = 40;//ms

        public DelayedThread() {
            super("------delayed--action---------");
            this.queue = new ConcurrentLinkedDeque<DelayedAction>();
            stop = false;
            setPriority(Thread.MAX_PRIORITY);
        }

        public void forceStop(){
            stop = true;
            queue.clear();
            queue = null;
        }

        @Override
        public void run() {
            while (!stop){
                if(!queue.isEmpty()){
                    DelayedAction last = queue.peekLast();
                    if(last != null){
                        long currTime = System.currentTimeMillis();
                        DelayedAction current;
                        while( (current = queue.removeFirst()) != null){
                            if(!current.tryExec(currTime)){
                                queue.offer(current);
                            }

                            if(current == last){
                                break;
                            }
                        }
                    }

                }

                try {
                    TimeUnit.MILLISECONDS.sleep(ZZ_TIME);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }

        public void checkinDelayAction(DelayedAction delayedAction) {
            this.queue.addLast(delayedAction);
        }
    }
}
